package com.dc.sbmq.service;

import com.dc.sbmq.config.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import com.rabbitmq.tools.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;

/**
 * 测试
 *
 * @author duchao
 */
@Slf4j
@Service
public class TestService {

    private static final CopyOnWriteArraySet<String> consumedMsgSet = new CopyOnWriteArraySet<>();
    private static final CyclicBarrier orderConsumeBarrier = new CyclicBarrier(2);

//    /**
//     * rabbitmq注解监听
//     * 需要开启注解监听@EnableRabbit
//     *
//     * @param message 消息
//     * @param body    消息体
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
//    public void testConsumeDirectQueue(Channel channel, Message message, @Payload String body) throws Exception {
//        try {
//            log.info("====消费者1-接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
//            log.info("消息id：{}",message.getMessageProperties().getMessageId());
//            log.info(body);
//            //模拟异常进入死信队列
//            //int i = 2 / new Random().nextInt(2);
//            //根据分布式消息id，进行消息幂等性处理，解决消息重复消费问题
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            log.info("消息手动确认接收(消费)");
//
//
//        } catch (Exception e) {
//            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
//            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
//            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
//        }
//    }



//    /**
//     * 测试消费重试机制，
//     * rabbitmq注解监听
//     * 需要开启注解监听@EnableRabbit
//     *
//     * @param message 消息
//     * @param body    消息体
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
//    public void testConsumeDirectQueueRetry(Channel channel, Message message, @Payload String body) throws Exception {
//        log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
//        log.info(body);
//        //模拟异常进入死信队列
//        int i = 2 / 0;
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//        log.info("消息手动确认接收(消费)");
//    }



//    /**
//     * 测试消息幂等性处理
//     * 解决消息重复消费问题
//     *
//     * rabbitmq注解监听
//     * 需要开启注解监听@EnableRabbit
//     *
//     * @param message 消息
//     * @param body    消息体
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
//    public void testIdempotentConsumeMsg(Channel channel, Message message, @Payload String body) throws Exception {
//        try {
//            //根据分布式消息id，进行消息幂等性处理，解决消息重复消费问题
//            if(consumedMsgSet.contains(message.getMessageProperties().getMessageId())){
//                log.error("不要重复消费消息");
//                return;
//            }
//
//            log.info("====消费者2-接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
//            log.info("消息id：{}",message.getMessageProperties().getMessageId());
//            log.info(body);
//            //模拟异常进入死信队列
//            int i = 2 / new Random().nextInt(2);
//            //根据分布式消息id，进行消息幂等性处理，解决消息重复消费问题
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            log.info("消息手动确认接收(消费)");
//
//            consumedMsgSet.add(message.getMessageProperties().getMessageId());
//        } catch (Exception e) {
//            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
//            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
//            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
//        }
//    }




    /**
     * 主题交换机绑定队列监听消费
     *
     * @param message 消息
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_TOPIC_QUEUE1)
    public void testConsumeTopicQueue1(Channel channel, Message message) throws IOException {
        log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_TOPIC_QUEUE1);
        log.info(new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("消息手动确认接收(消费)");
    }




    /**
     * 主题交换机绑定队列监听消费
     *
     * @param message 消息
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_TOPIC_QUEUE2)
    public void testConsumeTopicQueue2(Channel channel, Message message) throws IOException {
        log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_TOPIC_QUEUE2);
        log.info(new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("消息手动确认接收(消费)");
    }




    /**
     * 扇形交换机绑定队列监听消费
     *
     * @param message 消息
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_FANOUT_QUEUE1)
    public void testConsumeFanoutQueue1(Channel channel, Message message) throws IOException {
        log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_FANOUT_QUEUE1);
        log.info(new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("消息手动确认接收(消费)");
    }



    /**
     * 扇形交换机绑定队列监听消费
     *
     * @param message 消息
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_FANOUT_QUEUE2)
    public void testConsumeFanoutQueue2(Channel channel, Message message) throws IOException {
        log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_FANOUT_QUEUE2);
        log.info(new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("消息手动确认接收(消费)");
    }




//    /**
//     * 测试死信队列
//     *
//     * @param message 消息
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DEAD_LETTER_QUEUE1)
//    public void testConsumeDeadLetterQueue1(Channel channel, Message message) throws IOException {
//        //根据分布式消息id，进行消息幂等性处理，解决消息重复消费问题
//        if(consumedMsgSet.contains(message.getMessageProperties().getMessageId())){
//            log.error("不要重复消费消息");
//            return;
//        }
//        log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_DEAD_LETTER_QUEUE1);
//        log.info(new String(message.getBody()));
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//        log.info("消息手动确认接收(消费)");
//        consumedMsgSet.add(message.getMessageProperties().getMessageId());
//    }



//    /**
//     * 测试消息顺序消费
//     *
//     * 两个消费者
//     * 消息进入队列是先进先出按顺序出队列，但是多个消费者对消息消费不一定按顺序
//     * 解决：可以采用单消费者模式，一个队列只绑定一个消费者
//     *
//     * @param message 消息
//     * @param body    消息体
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
//    public void testOrderConsumeMsg1(Channel channel, Message message, @Payload String body) throws Exception {
//        try {
//            log.info("====消费者1-接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
//            log.info("收到消息：{}",body);
//            orderConsumeBarrier.await();
//            log.info("{}被消费",body);
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            log.info("消息手动确认接收(消费)");
//        } catch (Exception e) {
//            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
//            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
//            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
//        }
//    }
//
//    /**
//     * 测试消息顺序消费
//     *
//     * 两个消费者
//     * 消息进入队列是先进先出按顺序出队列，但是多个消费者对消息消费不一定按顺序
//     * 解决：可以采用单消费者模式，一个队列只绑定一个消费者
//     *
//     * @param message 消息
//     * @param body    消息体
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
//    public void testOrderConsumeMsg2(Channel channel, Message message, @Payload String body) throws Exception {
//        try {
//            log.info("====消费者2-接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
//            log.info("收到消息：{}",body);
//            orderConsumeBarrier.await();
//            log.info("{}被消费",body);
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            log.info("消息手动确认接收(消费)");
//        } catch (Exception e) {
//            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
//            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
//            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
//        }
//    }


//    /**
//     * 测试限流
//     *
//     * @param message 消息
//     * @param body    消息体
//     */
//    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
//    public void testLimitFlow(Channel channel, Message message, @Payload String body) throws Exception {
//        try {
//            // 表示不限制消息大小, 一次只处理一条消息, 限制只是当前消费者有效
//            channel.basicQos(0, 1, false);
//            log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
//            log.info("收到消息：{}",body);
//            Thread.sleep(1000);
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            log.info("消息手动确认接收(消费)");
//        } catch (Exception e) {
//            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
//            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
//            //channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
//        }
//    }

    /**
     * 分布式事务（消息发送）
     *
     * @param message 消息
     * @param body    消息体
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_DIRECT_QUEUE)
    public void testSendMsgWithTransaction(Channel channel, Message message, @Payload String body) throws Exception {
        try {
            log.info("====接收到{}队列消息=====", RabbitMQConfig.DEMO_DIRECT_QUEUE);
            log.info("收到消息：{}",body);
            //channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            //log.info("消息手动确认接收(消费)");
        } catch (Exception e) {
            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
            //channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    /**
     * 延迟队列
     *
     * @param message 消息
     * @param body    消息体
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_DELAY_OUT_QUEUE)
    public void testDelayQueue(Channel channel, Message message, @Payload String body) throws Exception {
        try {
            log.info("====接收到{}模拟延迟队列消息=====", RabbitMQConfig.DEMO_DELAY_OUT_QUEUE);
            log.info("收到消息：{}",body);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //接收确认失败，建议为队列设置死信队列。当出现异常时，由死信队列监听后再进行处理。
            //以下操作，如果队列已经绑定了死信队列，消息会发送到死信队列
            //channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}
